/*-------------------------------------------------------------------------
 *
 * slotfuncs.c
 *	   复制插槽的支持函数
 *
 * Copyright (c) 2012-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/replication/slotfuncs.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/htup_details.h"
#include "access/xlog_internal.h"
#include "access/xlogrecovery.h"
#include "access/xlogutils.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "replication/decode.h"
#include "replication/logical.h"
#include "replication/slot.h"
#include "utils/builtins.h"
#include "utils/inval.h"
#include "utils/pg_lsn.h"
#include "utils/resowner.h"

/*
 * 创建一个新的物理复制插槽的辅助函数，接受给定参数。注意，此函数不会释放创建的插槽。
 *
 * 如果 restart_lsn 是一个有效值，我们将使用它而不使用 WAL 保留例程。因此调用者必须保证 WAL 可用。
 */
static void fc_create_physical_replication_slot(char *fc_name, bool fc_immediately_reserve,
#ifdef FDD
								 bool fc_failover,
#endif
								 bool fc_temporary, XLogRecPtr fc_restart_lsn)
{
	Assert(!MyReplicationSlot);

	/* 获取复制插槽，这将检查冲突名称 */
	ReplicationSlotCreate(fc_name, false,
#ifdef FDD
						  fc_temporary ? RS_TEMPORARY : RS_PERSISTENT, false, fc_failover);
#else
						  fc_temporary ? RS_TEMPORARY : RS_PERSISTENT, false);
#endif

	if (fc_immediately_reserve)
	{
		/* 根据用户要求保留 WAL */
		if (XLogRecPtrIsInvalid(fc_restart_lsn))
			ReplicationSlotReserveWal();
		else
			MyReplicationSlot->data.restart_lsn = fc_restart_lsn;

		/* 将此插槽写入磁盘 */
		ReplicationSlotMarkDirty();
		ReplicationSlotSave();
	}
}

/*
 * 创建一个新的逻辑（流复制）复制插槽的 SQL 函数。
 */
Datum pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
{
	Name		fc_name = PG_GETARG_NAME(0);
	bool		fc_immediately_reserve = PG_GETARG_BOOL(1);
	bool		fc_temporary = PG_GETARG_BOOL(2);
#ifdef FDD
	bool		fc_failover = false;
#endif
	Datum		fc_values[2];
	bool		fc_nulls[2];
	TupleDesc	fc_tupdesc;
	HeapTuple	fc_tuple;
	Datum		fc_result;

#ifdef FDD
	/*
	 * 此函数可以使用 pg_proc 中的标准签名调用，或者使用来自故障转移插槽扩展的新签名。它必须能够处理两者。
	 */
	if (PG_NARGS() == 4)
		fc_failover = PG_GETARG_BOOL(3);
#endif

	if (get_call_result_type(fcinfo, NULL, &fc_tupdesc) != TYPEFUNC_COMPOSITE)
		elog(ERROR, "return type must be a row type");

	CheckSlotPermissions();

	CheckSlotRequirements();

	fc_create_physical_replication_slot(NameStr(*fc_name),
									 fc_immediately_reserve,
#ifdef FDD
									 fc_failover,
#endif
									 fc_temporary,
									 InvalidXLogRecPtr);

	fc_values[0] = NameGetDatum(&MyReplicationSlot->data.name);
	fc_nulls[0] = false;

	if (fc_immediately_reserve)
	{
		fc_values[1] = LSNGetDatum(MyReplicationSlot->data.restart_lsn);
		fc_nulls[1] = false;
	}
	else
		fc_nulls[1] = true;

	fc_tuple = heap_form_tuple(fc_tupdesc, fc_values, fc_nulls);
	fc_result = HeapTupleGetDatum(fc_tuple);

	ReplicationSlotRelease();

	PG_RETURN_DATUM(fc_result);
}


/*
 * 创建一个新的逻辑复制插槽的辅助函数，接受给定参数。注意，此函数不会释放创建的插槽。
 *
 * 当 find_startpoint 为 false 时，插槽的 confirmed_flush 不会被设置；确保其设置为某个合理值是调用者的责任。
 */
static void fc_create_logical_replication_slot(char *fc_name, char *fc_plugin,
								bool fc_temporary, bool fc_two_phase,
#ifdef FDD
								bool fc_failover,
#endif
								XLogRecPtr fc_restart_lsn,
								bool fc_find_startpoint)
{
	LogicalDecodingContext *fc_ctx = NULL;

	Assert(!MyReplicationSlot);

	/*
	 * 获取一个逻辑解码插槽，这将检查冲突名称。
	 * 最初将持久插槽创建为暂时的 - 这使我们能够在初始化过程中很好地处理错误，因为如果此事务失败，它将被丢弃。我们将在最后将其设为持久。临时插槽可以从一开始就作为临时插槽创建，因为在错误时也会被丢弃。
	 */
	ReplicationSlotCreate(fc_name, true,
#ifdef FDD
						  fc_temporary ? RS_TEMPORARY : RS_EPHEMERAL, fc_two_phase, fc_failover);
#else
						  fc_temporary ? RS_TEMPORARY : RS_EPHEMERAL, fc_two_phase);
#endif

	/*
	 * 创建逻辑解码上下文以查找起始点，或者如果我们不需要它，则 1）提升插槽的 restart_lsn 和 xmin 2）检查插件的有效性。
	 *
	 * 注意：当 !find_startpoint 时，这仍然很重要，因为此时输出插件被验证。
	 */
	fc_ctx = CreateInitDecodingContext(fc_plugin, NIL,
									false,	/* 仅目录是可以的 */
									fc_restart_lsn,
									XL_ROUTINE(.page_read = read_local_xlog_page,
											   .segment_open = wal_segment_open,
											   .segment_close = wal_segment_close),
									NULL, NULL, NULL);

	/*
	 * 如果调用者需要我们确定解码起始点，现在就这样做。
	 * 这可能需要一些时间。
	 */
	if (fc_find_startpoint)
		DecodingContextFindStartpoint(fc_ctx);

	/* 不再需要解码上下文 */
	FreeDecodingContext(fc_ctx);
}

/*
 * 创建一个新的逻辑复制插槽的 SQL 函数。
 */
Datum pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
{
	Name		fc_name = PG_GETARG_NAME(0);
	Name		fc_plugin = PG_GETARG_NAME(1);
	bool		fc_temporary = PG_GETARG_BOOL(2);
	bool		fc_two_phase = PG_GETARG_BOOL(3);
#ifdef FDD
	bool		fc_failover = false;
#endif

	Datum		fc_result;
	TupleDesc	fc_tupdesc;
	HeapTuple	fc_tuple;
	Datum		fc_values[2];
	bool		fc_nulls[2];

#ifdef FDD
	/*
	 * 此函数可以使用 pg_proc 中的标准签名调用，或者使用来自故障转移插槽扩展的新签名。它必须能够处理两者。
	 */
	if (PG_NARGS() == 5)
	{
		fc_failover = PG_GETARG_BOOL(4);
	}
#endif

	if (get_call_result_type(fcinfo, NULL, &fc_tupdesc) != TYPEFUNC_COMPOSITE)
		elog(ERROR, "return type must be a row type");

	CheckSlotPermissions();

	CheckLogicalDecodingRequirements();

	fc_create_logical_replication_slot(NameStr(*fc_name),
									NameStr(*fc_plugin),
									fc_temporary,
									fc_two_phase,
#ifdef FDD
									fc_failover,
#endif
									InvalidXLogRecPtr,
									true);

	fc_values[0] = NameGetDatum(&MyReplicationSlot->data.name);
	fc_values[1] = LSNGetDatum(MyReplicationSlot->data.confirmed_flush);

	memset(fc_nulls, 0, sizeof(fc_nulls));

	fc_tuple = heap_form_tuple(fc_tupdesc, fc_values, fc_nulls);
	fc_result = HeapTupleGetDatum(fc_tuple);

	/* 好的，插槽现在完全创建，如果需要，请将其标记为持久 */
	if (!fc_temporary)
		ReplicationSlotPersist();
	ReplicationSlotRelease();

	PG_RETURN_DATUM(fc_result);
}


/*
 * 删除复制插槽的 SQL 函数。
 */
Datum pg_drop_replication_slot(PG_FUNCTION_ARGS)
{
	Name		fc_name = PG_GETARG_NAME(0);

	CheckSlotPermissions();

	CheckSlotRequirements();

	ReplicationSlotDrop(NameStr(*fc_name), true);

	PG_RETURN_VOID();
}

/*
 * pg_get_replication_slots - 显示活动复制插槽的 SQL SRF。
 */
Datum pg_get_replication_slots(PG_FUNCTION_ARGS)
{
#ifdef FDD
#define PG_GET_REPLICATION_SLOTS_COLS 15
#else
#define PG_GET_REPLICATION_SLOTS_COLS 14
#endif
	ReturnSetInfo *fc_rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
	XLogRecPtr	fc_currlsn;
	int			fc_slotno;

	/*
	 * 我们不需要任何特殊权限来查看此函数的数据，因为没有东西是敏感的。最关键的是插槽名称，插槽名称不应包含任何特别敏感的内容。
	 */

	InitMaterializedSRF(fcinfo, 0);

	fc_currlsn = GetXLogWriteRecPtr();

	LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
	for (fc_slotno = 0; fc_slotno < max_replication_slots; fc_slotno++)
	{
		ReplicationSlot *fc_slot = &ReplicationSlotCtl->replication_slots[fc_slotno];
		ReplicationSlot fc_slot_contents;
		Datum		fc_values[PG_GET_REPLICATION_SLOTS_COLS];
		bool		fc_nulls[PG_GET_REPLICATION_SLOTS_COLS];
		WALAvailability fc_walstate;
		int			fc_i;

		if (!fc_slot->in_use)
			continue;

		/* 在保持自旋锁的同时复制插槽内容，然后从容检查 */
		SpinLockAcquire(&fc_slot->mutex);
		fc_slot_contents = *fc_slot;
		SpinLockRelease(&fc_slot->mutex);

		memset(fc_values, 0, sizeof(fc_values));
		memset(fc_nulls, 0, sizeof(fc_nulls));

		fc_i = 0;
		fc_values[fc_i++] = NameGetDatum(&fc_slot_contents.data.name);

		if (fc_slot_contents.data.database == InvalidOid)
			fc_nulls[fc_i++] = true;
		else
			fc_values[fc_i++] = NameGetDatum(&fc_slot_contents.data.plugin);

		if (fc_slot_contents.data.database == InvalidOid)
			fc_values[fc_i++] = CStringGetTextDatum("physical");
		else
			fc_values[fc_i++] = CStringGetTextDatum("logical");

		if (fc_slot_contents.data.database == InvalidOid)
			fc_nulls[fc_i++] = true;
		else
			fc_values[fc_i++] = ObjectIdGetDatum(fc_slot_contents.data.database);

		fc_values[fc_i++] = BoolGetDatum(fc_slot_contents.data.persistency == RS_TEMPORARY);
		fc_values[fc_i++] = BoolGetDatum(fc_slot_contents.active_pid != 0);

		if (fc_slot_contents.active_pid != 0)
			fc_values[fc_i++] = Int32GetDatum(fc_slot_contents.active_pid);
		else
			fc_nulls[fc_i++] = true;

		if (fc_slot_contents.data.xmin != InvalidTransactionId)
			fc_values[fc_i++] = TransactionIdGetDatum(fc_slot_contents.data.xmin);
		else
			fc_nulls[fc_i++] = true;

		if (fc_slot_contents.data.catalog_xmin != InvalidTransactionId)
			fc_values[fc_i++] = TransactionIdGetDatum(fc_slot_contents.data.catalog_xmin);
		else
			fc_nulls[fc_i++] = true;

		if (fc_slot_contents.data.restart_lsn != InvalidXLogRecPtr)
			fc_values[fc_i++] = LSNGetDatum(fc_slot_contents.data.restart_lsn);
		else
			fc_nulls[fc_i++] = true;

		if (fc_slot_contents.data.confirmed_flush != InvalidXLogRecPtr)
			fc_values[fc_i++] = LSNGetDatum(fc_slot_contents.data.confirmed_flush);
		else
			fc_nulls[fc_i++] = true;

		/*
		 * 如果 invalidated_at 是有效的而 restart_lsn 是无效的，我们可以确定插槽已被作废。否则，测试从 restart_lsn 的可用性。
		 */
		if (XLogRecPtrIsInvalid(fc_slot_contents.data.restart_lsn) &&
			!XLogRecPtrIsInvalid(fc_slot_contents.data.invalidated_at))
			fc_walstate = WALAVAIL_REMOVED;
		else
			fc_walstate = GetWALAvailability(fc_slot_contents.data.restart_lsn);

		switch (fc_walstate)
		{
			case WALAVAIL_INVALID_LSN:
				fc_nulls[fc_i++] = true;
				break;

			case WALAVAIL_RESERVED:
				fc_values[fc_i++] = CStringGetTextDatum("reserved");
				break;

			case WALAVAIL_EXTENDED:
				fc_values[fc_i++] = CStringGetTextDatum("extended");
				break;

			case WALAVAIL_UNRESERVED:
				fc_values[fc_i++] = CStringGetTextDatum("unreserved");
				break;

			case WALAVAIL_REMOVED:

				/*
				 * 如果我们读取restart_lsn的时间太久，也许那个文件
				 * 现在已经被移除了。然而，walsender可能已经前进到
				 * 另一个文件，即便我们查看的时候在另一个文件。如果
				 * checkpointer发信号让进程终止，那肯定会丢失；但如果
				 * 一个进程仍然存活，那么“不保留”的说法似乎更为合适。
				 *
				 * 如果我们确实更改了它，需保存下面的safe_wal_size状态。
				 */
				if (!XLogRecPtrIsInvalid(fc_slot_contents.data.restart_lsn))
				{
					int			fc_pid;

					SpinLockAcquire(&fc_slot->mutex);
					fc_pid = fc_slot->active_pid;
					fc_slot_contents.data.restart_lsn = fc_slot->data.restart_lsn;
					SpinLockRelease(&fc_slot->mutex);
					if (fc_pid != 0)
					{
						fc_values[fc_i++] = CStringGetTextDatum("unreserved");
						fc_walstate = WALAVAIL_UNRESERVED;
						break;
					}
				}
				fc_values[fc_i++] = CStringGetTextDatum("lost");
				break;
		}

		/*
		 * safe_wal_size仅针对尚未丢失的槽进行计算，
		 * 并且仅在配置了最大大小的情况下。
		 */
		if (fc_walstate == WALAVAIL_REMOVED || max_slot_wal_keep_size_mb < 0)
			fc_nulls[fc_i++] = true;
		else
		{
			XLogSegNo	fc_targetSeg;
			uint64		fc_slotKeepSegs;
			uint64		fc_keepSegs;
			XLogSegNo	fc_failSeg;
			XLogRecPtr	fc_failLSN;

			XLByteToSeg(fc_slot_contents.data.restart_lsn, fc_targetSeg, wal_segment_size);

			/* 确定槽可以保持多少段 */
			fc_slotKeepSegs = XLogMBVarToSegs(max_slot_wal_keep_size_mb, wal_segment_size);
			/* wal_keep_size同上 */
			fc_keepSegs = XLogMBVarToSegs(wal_keep_size_mb, wal_segment_size);

			/* 如果currpos达到failLSN，我们就失去我们的段 */
			fc_failSeg = fc_targetSeg + Max(fc_slotKeepSegs, fc_keepSegs) + 1;
			XLogSegNoOffsetToRecPtr(fc_failSeg, 0, wal_segment_size, fc_failLSN);

			fc_values[fc_i++] = Int64GetDatum(fc_failLSN - fc_currlsn);
		}

		fc_values[fc_i++] = BoolGetDatum(fc_slot_contents.data.two_phase);

#ifdef FDD
		/*
		 * 此函数可以使用pg_proc中的标准签名进行调用，
		 * 在这种情况下，它不期望额外的'failover'列，
		 * 或者使用我们扩展中的新签名，那里有空间。我们可以
		 * 使用期望的tupledesc区分这两种情况，但实际上我们
		 * 不需要这样做 - 在结果集数组中分配额外空间并填充
		 * 是无害的。如果不期望，它将被忽略，但这意味着
		 * failover字段必须在结果中最后。
		 */
		fc_values[fc_i++] = BoolGetDatum(fc_slot_contents.data.failover);
#endif

		Assert(fc_i == PG_GET_REPLICATION_SLOTS_COLS);

		tuplestore_putvalues(fc_rsinfo->setResult, fc_rsinfo->setDesc,
							 fc_values, fc_nulls);
	}

	LWLockRelease(ReplicationSlotControlLock);

	return (Datum) 0;
}

/*
 * 用于推进物理复制槽的辅助函数。
 *
 * 要移动的LSN位置仅与槽的restart_lsn进行比较，
 * 知道比这更老的位置将会因连续的检查点而被移除。
 */
static XLogRecPtr fc_pg_physical_replication_slot_advance(XLogRecPtr fc_moveto)
{
	XLogRecPtr	fc_startlsn = MyReplicationSlot->data.restart_lsn;
	XLogRecPtr	fc_retlsn = fc_startlsn;

	Assert(fc_moveto != InvalidXLogRecPtr);

	if (fc_startlsn < fc_moveto)
	{
		SpinLockAcquire(&MyReplicationSlot->mutex);
		MyReplicationSlot->data.restart_lsn = fc_moveto;
		SpinLockRelease(&MyReplicationSlot->mutex);
		fc_retlsn = fc_moveto;

		/*
		 * 使槽脏，以便在下一个检查点时写出。请注意，
		 * 已推进的LSN位置在崩溃时可能仍会丢失，
		 * 但这使得在干净关闭后数据一致。
		 */
		ReplicationSlotMarkDirty();
	}

	return fc_retlsn;
}

/*
 * 用于推进逻辑复制槽的辅助函数。
 *
 * 槽的restart_lsn被用作读取记录的起始点，而
 * confirmed_flush被用作解码上下文的基点。
 *
 * 我们不能仅仅使用LogicalConfirmReceivedLocation来更新
 * confirmed_flush，因为我们需要消化WAL以推进restart_lsn，
 * 允许回收WAL并删除旧的目录元组。由于解码以快速前进
 * 模式进行，因此不会生成任何更改。
 */
static XLogRecPtr fc_pg_logical_replication_slot_advance(XLogRecPtr fc_moveto)
{
	LogicalDecodingContext *fc_ctx;
	ResourceOwner fc_old_resowner = CurrentResourceOwner;
	XLogRecPtr	fc_retlsn;

	Assert(fc_moveto != InvalidXLogRecPtr);

	PG_TRY();
	{
		/*
		 * 在快速前进模式下创建我们的解码上下文，并将start_lsn
		 * 作为InvalidXLogRecPtr传递，这样我们就可以从我的槽的
		 * confirmed_flush开始处理。
		 */
		fc_ctx = CreateDecodingContext(InvalidXLogRecPtr,
									NIL,
									true,	/* 快速前进 */
									XL_ROUTINE(.page_read = read_local_xlog_page,
											   .segment_open = wal_segment_open,
											   .segment_close = wal_segment_close),
									NULL, NULL, NULL);

		/*
		 * 从槽的restart_lsn开始读取，我们知道它指向
		 * 有效记录。
		 */
		XLogBeginRead(fc_ctx->reader, MyReplicationSlot->data.restart_lsn);

		/* 无效的非时间旅行条目 */
		InvalidateSystemCaches();

		/* 至少解码一条记录，直到我们没有记录为止 */
		while (fc_ctx->reader->EndRecPtr < fc_moveto)
		{
			char	   *fc_errm = NULL;
			XLogRecord *fc_record;

			/*
			 * 读取记录。在快速前进模式下不会生成更改，
			 * 但snapbuilder/槽状态会得到正确更新。
			 */
			fc_record = XLogReadRecord(fc_ctx->reader, &fc_errm);
			if (fc_errm)
				elog(ERROR, "could not find record while advancing replication slot: %s",
					 fc_errm);

			/*
			 * 处理记录。在快速转发模式下，存储级别的更改将被忽略，
			 * 但其他模块（如 snapbuilder）可能仍需要进行关键更新。
			 */
			if (fc_record)
				LogicalDecodingProcessRecord(fc_ctx, fc_ctx->reader);

			/* 一旦达到请求的目标则停止 */
			if (fc_moveto <= fc_ctx->reader->EndRecPtr)
				break;

			CHECK_FOR_INTERRUPTS();
		}

		/*
		 * 逻辑解码可能在事务管理期间覆盖了 CurrentResourceOwner，
		 * 因此需要恢复执行者的值。（这是一种应急措施，但现在不值得清理。）
		 */
		CurrentResourceOwner = fc_old_resowner;

		if (fc_ctx->reader->EndRecPtr != InvalidXLogRecPtr)
		{
			LogicalConfirmReceivedLocation(fc_moveto);

			/*
			 * 如果仅仅是 confirmed_flush LSN 改变，插槽不会被上述标记为脏。
			 * walsender 接口的调用者预计会跟踪他们自己的进度，
			 * 不需要它被写出。但 SQL 接口的用户不能指定自己的起始位置，
			 * 而且他们跟踪进度更困难，因此我们应该更加努力地为他们保存进度。
			 *
			 * 将插槽标记为脏，以便在下一个检查点写出。所前进的 LSN
			 * 位置可能在崩溃中丢失，但这使得数据在干净关闭后保持一致。
			 */
			ReplicationSlotMarkDirty();
		}

		fc_retlsn = MyReplicationSlot->data.confirmed_flush;

		/* 释放上下文，调用关闭回调 */
		FreeDecodingContext(fc_ctx);

		InvalidateSystemCaches();
	}
	PG_CATCH();
	{
		/* 清除所有时间旅行条目 */
		InvalidateSystemCaches();

		PG_RE_THROW();
	}
	PG_END_TRY();

	return fc_retlsn;
}

/*
 * 用于在复制插槽中移动位置的 SQL 函数。
 */
Datum pg_replication_slot_advance(PG_FUNCTION_ARGS)
{
	Name		fc_slotname = PG_GETARG_NAME(0);
	XLogRecPtr	fc_moveto = PG_GETARG_LSN(1);
	XLogRecPtr	fc_endlsn;
	XLogRecPtr	fc_minlsn;
	TupleDesc	fc_tupdesc;
	Datum		fc_values[2];
	bool		fc_nulls[2];
	HeapTuple	fc_tuple;
	Datum		fc_result;

	Assert(!MyReplicationSlot);

	CheckSlotPermissions();

	if (XLogRecPtrIsInvalid(fc_moveto))
		ereport(ERROR,
				(errmsg("invalid target WAL LSN")));

	/* 为我们的结果类型构建一个元组描述符 */
	if (get_call_result_type(fcinfo, NULL, &fc_tupdesc) != TYPEFUNC_COMPOSITE)
		elog(ERROR, "return type must be a row type");

	/*
	 * 我们不能将插槽移动超过已被冲刷/重放的内容，因此相应地限制目标位置。
	 */
	if (!RecoveryInProgress())
		fc_moveto = Min(fc_moveto, GetFlushRecPtr(NULL));
	else
		fc_moveto = Min(fc_moveto, GetXLogReplayRecPtr(NULL));

	/* 获取插槽以便我们“拥有”它 */
	ReplicationSlotAcquire(NameStr(*fc_slotname), true);

	/* 一个 restart_lsn 从未被保留的插槽无法被推进 */
	if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn))
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("replication slot \"%s\" cannot be advanced",
						NameStr(*fc_slotname)),
				 errdetail("This slot has never previously reserved WAL, or it has been invalidated.")));

	/*
	 * 检查插槽是否没有向后移动。物理插槽仅简单依赖 restart_lsn
	 * 作为最小点，而逻辑插槽已确认消费至 confirmed_flush，
	 * 这意味着在这两种情况下，旧于此的数据不再可用。
	 */
	if (OidIsValid(MyReplicationSlot->data.database))
		fc_minlsn = MyReplicationSlot->data.confirmed_flush;
	else
		fc_minlsn = MyReplicationSlot->data.restart_lsn;

	if (fc_moveto < fc_minlsn)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("cannot advance replication slot to %X/%X, minimum is %X/%X",
						LSN_FORMAT_ARGS(fc_moveto), LSN_FORMAT_ARGS(fc_minlsn))));

	/* 根据插槽类型进行实际插槽更新 */
	if (OidIsValid(MyReplicationSlot->data.database))
		fc_endlsn = fc_pg_logical_replication_slot_advance(fc_moveto);
	else
		fc_endlsn = fc_pg_physical_replication_slot_advance(fc_moveto);

	fc_values[0] = NameGetDatum(&MyReplicationSlot->data.name);
	fc_nulls[0] = false;

	/*
	 * 重新计算所有插槽的最小 LSN 和 xmin，以便根据
	 * 潜在的推进进行调整。
	 */
	ReplicationSlotsComputeRequiredXmin(false);
	ReplicationSlotsComputeRequiredLSN();

	ReplicationSlotRelease();

	/* 返回达到的位置。 */
	fc_values[1] = LSNGetDatum(fc_endlsn);
	fc_nulls[1] = false;

	fc_tuple = heap_form_tuple(fc_tupdesc, fc_values, fc_nulls);
	fc_result = HeapTupleGetDatum(fc_tuple);

	PG_RETURN_DATUM(fc_result);
}

/*
 * 复制复制插槽的辅助函数。
 */
static Datum fc_copy_replication_slot(FunctionCallInfo fcinfo, bool fc_logical_slot)
{
	Name		fc_src_name = PG_GETARG_NAME(0);
	Name		fc_dst_name = PG_GETARG_NAME(1);
	ReplicationSlot *fc_src = NULL;
	ReplicationSlot fc_first_slot_contents;
	ReplicationSlot fc_second_slot_contents;
	XLogRecPtr	fc_src_restart_lsn;
	bool		fc_src_islogical;
	bool		fc_temporary;
	char	   *fc_plugin;
	Datum		fc_values[2];
	bool		fc_nulls[2];
	Datum		fc_result;
	TupleDesc	fc_tupdesc;
	HeapTuple	fc_tuple;

	if (get_call_result_type(fcinfo, NULL, &fc_tupdesc) != TYPEFUNC_COMPOSITE)
		elog(ERROR, "return type must be a row type");

	CheckSlotPermissions();

	if (fc_logical_slot)
		CheckLogicalDecodingRequirements();
	else
		CheckSlotRequirements();

	LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);

	/*
	 * 我们需要防止源插槽的保留 WAL 被删除，
	 * 但我们不想锁定该插槽过长时间，同时它可以在此期间进行推进。
	 * 所以获取源插槽的数据，并使用其 restart_lsn 创建一个新插槽。
	 * 之后我们再次锁定源插槽并验证我们复制的数据（名称，类型）没有
	 * 发生不兼容的变化。一旦创建了新插槽，便不会发生不方便的 WAL 删除
	 * ——但由于 WAL 删除可能在我们成功创建新插槽之前已经发生，
	 * 因此我们在第二次锁定它时将新插槽的 restart_lsn 推进到源插槽的更新
	 * restart_lsn。
	 */
	for (int fc_i = 0; fc_i < max_replication_slots; fc_i++)
	{
		ReplicationSlot *fc_s = &ReplicationSlotCtl->replication_slots[fc_i];

		if (fc_s->in_use && strcmp(NameStr(fc_s->data.name), NameStr(*fc_src_name)) == 0)
		{
			/* 在持有自旋锁的同时复制插槽内容 */
			SpinLockAcquire(&fc_s->mutex);
			fc_first_slot_contents = *fc_s;
			SpinLockRelease(&fc_s->mutex);
			fc_src = fc_s;
			break;
		}
	}

	LWLockRelease(ReplicationSlotControlLock);

	if (fc_src == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("replication slot \"%s\" does not exist", NameStr(*fc_src_name))));

	fc_src_islogical = SlotIsLogical(&fc_first_slot_contents);
	fc_src_restart_lsn = fc_first_slot_contents.data.restart_lsn;
	fc_temporary = (fc_first_slot_contents.data.persistency == RS_TEMPORARY);
	fc_plugin = fc_logical_slot ? NameStr(fc_first_slot_contents.data.plugin) : NULL;

	/* 检查复制插槽的类型 */
	if (fc_src_islogical != fc_logical_slot)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 fc_src_islogical ?
				 errmsg("cannot copy physical replication slot \"%s\" as a logical replication slot",
						NameStr(*fc_src_name)) :
				 errmsg("cannot copy logical replication slot \"%s\" as a physical replication slot",
						NameStr(*fc_src_name))));

	/* 复制非保留插槽没有意义 */
	if (XLogRecPtrIsInvalid(fc_src_restart_lsn))
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("cannot copy a replication slot that doesn't reserve WAL")));

	/* 从可选参数中覆盖参数 */
	if (PG_NARGS() >= 3)
		fc_temporary = PG_GETARG_BOOL(2);
	if (PG_NARGS() >= 4)
	{
		Assert(fc_logical_slot);
		fc_plugin = NameStr(*(PG_GETARG_NAME(3)));
	}

	/* 创建新的槽并获取它 */
	if (fc_logical_slot)
	{
		/*
		 * 我们不能尝试读取WAL，因为我们还没有预留它 --
		 * 因此传递find_startpoint为false。confirmed_flush将在
		 * 下面通过从源槽复制来设置。
		 */
		fc_create_logical_replication_slot(NameStr(*fc_dst_name),
										fc_plugin,
										fc_temporary,
										false,
#ifdef FDD
										false,
#endif
										fc_src_restart_lsn,
										false);
	}
	else
		fc_create_physical_replication_slot(NameStr(*fc_dst_name),
										 true,
#ifdef FDD
										 false,
#endif
										 fc_temporary,
										 fc_src_restart_lsn);

	/*
	 * 将目标槽更新为源槽的当前值；
	 * 重新检查源槽是否仍然是我们之前看到的那个。
	 */
	{
		TransactionId fc_copy_effective_xmin;
		TransactionId fc_copy_effective_catalog_xmin;
		TransactionId fc_copy_xmin;
		TransactionId fc_copy_catalog_xmin;
		XLogRecPtr	fc_copy_restart_lsn;
		XLogRecPtr	fc_copy_confirmed_flush;
		bool		fc_copy_islogical;
		char	   *fc_copy_name;

		/* 再次复制源槽的数据 */
		SpinLockAcquire(&fc_src->mutex);
		fc_second_slot_contents = *fc_src;
		SpinLockRelease(&fc_src->mutex);

		fc_copy_effective_xmin = fc_second_slot_contents.effective_xmin;
		fc_copy_effective_catalog_xmin = fc_second_slot_contents.effective_catalog_xmin;

		fc_copy_xmin = fc_second_slot_contents.data.xmin;
		fc_copy_catalog_xmin = fc_second_slot_contents.data.catalog_xmin;
		fc_copy_restart_lsn = fc_second_slot_contents.data.restart_lsn;
		fc_copy_confirmed_flush = fc_second_slot_contents.data.confirmed_flush;

		/* 用于存在性检查 */
		fc_copy_name = NameStr(fc_second_slot_contents.data.name);
		fc_copy_islogical = SlotIsLogical(&fc_second_slot_contents);

		/*
		 * 检查源槽是否仍然存在且有效。如果复制槽的类型或名称发生
		 * 更改，或者restart_lsn无效或回退，则我们认为它无效。
		 * （如果源槽被删除并在安装期间从旧槽复制，则
		 * restart_lsn可能会回退。）
		 *
		 * 由于出错将释放并删除目标槽，因此我们
		 * 在这里不需要释放它。
		 */
		if (fc_copy_restart_lsn < fc_src_restart_lsn ||
			fc_src_islogical != fc_copy_islogical ||
			strcmp(fc_copy_name, NameStr(*fc_src_name)) != 0)
			ereport(ERROR,
					(errmsg("could not copy replication slot \"%s\"",
							NameStr(*fc_src_name)),
					 errdetail("The source replication slot was modified incompatibly during the copy operation.")));

		/* 源槽必须具有一致的快照 */
		if (fc_src_islogical && XLogRecPtrIsInvalid(fc_copy_confirmed_flush))
			ereport(ERROR,
					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
					 errmsg("cannot copy unfinished logical replication slot \"%s\"",
							NameStr(*fc_src_name)),
					 errhint("Retry when the source replication slot's confirmed_flush_lsn is valid.")));

		/* 再次安装复制的值 */
		SpinLockAcquire(&MyReplicationSlot->mutex);
		MyReplicationSlot->effective_xmin = fc_copy_effective_xmin;
		MyReplicationSlot->effective_catalog_xmin = fc_copy_effective_catalog_xmin;

		MyReplicationSlot->data.xmin = fc_copy_xmin;
		MyReplicationSlot->data.catalog_xmin = fc_copy_catalog_xmin;
		MyReplicationSlot->data.restart_lsn = fc_copy_restart_lsn;
		MyReplicationSlot->data.confirmed_flush = fc_copy_confirmed_flush;
		SpinLockRelease(&MyReplicationSlot->mutex);

		ReplicationSlotMarkDirty();
		ReplicationSlotsComputeRequiredXmin(false);
		ReplicationSlotsComputeRequiredLSN();
		ReplicationSlotSave();

#ifdef USE_ASSERT_CHECKING
		/* 检查restart_lsn是否可用 */
		{
			XLogSegNo	fc_segno;

			XLByteToSeg(fc_copy_restart_lsn, fc_segno, wal_segment_size);
			Assert(XLogGetLastRemovedSegno() < fc_segno);
		}
#endif
	}

	/* 目标槽完全创建，如果需要则标记为持久性 */
	if (fc_logical_slot && !fc_temporary)
		ReplicationSlotPersist();

	/* 所有工作完成。 设置返回值 */
	fc_values[0] = NameGetDatum(fc_dst_name);
	fc_nulls[0] = false;
	if (!XLogRecPtrIsInvalid(MyReplicationSlot->data.confirmed_flush))
	{
		fc_values[1] = LSNGetDatum(MyReplicationSlot->data.confirmed_flush);
		fc_nulls[1] = false;
	}
	else
		fc_nulls[1] = true;

	fc_tuple = heap_form_tuple(fc_tupdesc, fc_values, fc_nulls);
	fc_result = HeapTupleGetDatum(fc_tuple);

	ReplicationSlotRelease();

	PG_RETURN_DATUM(fc_result);
}

/* 下面的包装器都是为了满足opr_sanity */
Datum pg_copy_logical_replication_slot_a(PG_FUNCTION_ARGS)
{
	return fc_copy_replication_slot(fcinfo, true);
}

Datum pg_copy_logical_replication_slot_b(PG_FUNCTION_ARGS)
{
	return fc_copy_replication_slot(fcinfo, true);
}

Datum pg_copy_logical_replication_slot_c(PG_FUNCTION_ARGS)
{
	return fc_copy_replication_slot(fcinfo, true);
}

Datum pg_copy_physical_replication_slot_a(PG_FUNCTION_ARGS)
{
	return fc_copy_replication_slot(fcinfo, false);
}

Datum pg_copy_physical_replication_slot_b(PG_FUNCTION_ARGS)
{
	return fc_copy_replication_slot(fcinfo, false);
}
